/****************************************************************************
 * Copyright 2009 Huascar A. Sanchez                                        *
 *                                                                          *
 * Licensed under the Apache License, Version 2.0 (the "License");          *
 * you may not use this file except in compliance with the License.         *
 * You may obtain a copy of the License at                                  *
 *                                                                          *
 *     http://www.apache.org/licenses/LICENSE-2.0                           *
 *                                                                          *
 * Unless required by applicable law or agreed to in writing, software      *
 * distributed under the License is distributed on an "AS IS" BASIS,        *
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
 * See the License for the specific language governing permissions and      *
 * limitations under the License.                                           *
 ****************************************************************************/
package com.gotobject.greatq.circuitbreaker;

import com.gotobject.filter.internal.ToStrings;

import java.lang.reflect.Constructor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import static com.gotobject.filter.internal.Pair.of;
import static com.gotobject.filter.internal.ToStrings.str;

/**
 * @author huascar.sanchez@gmail.com (Huascar A. Sanchez)
 */
public class TunableCircuitBreaker<T extends Exception> implements CircuitBreaker<T>, Monitored {
    private static final long FAILED_CALL_THRESHOLD = 1L;
    private static final long RETRY_THRESHOLD       = 10000L;
    private static final long NO_CLOSED_YET         = -1L;


    // # of failed calls since the cb closed.
    private AtomicLong                  failedCalls         = new AtomicLong();
    // # of calls blocked by the cb.
    private AtomicLong                  blockedCalls        = new AtomicLong();
    // cb's current status
    private AtomicReference<Status>     status              = new AtomicReference<Status>(Status.CLOSED);
    // last registered exception if the cb's status == Open. Otherwise, it should be NULL.
    private AtomicReference<T>          lastException       = new AtomicReference<T>();
    // time at which the circuit breaker may retry letting an operation to be called.
    private AtomicLong                  nextTryTime         = new AtomicLong();
    // time at which the cb was able to reset itself.
    private AtomicLong                  avgRecoveryTime     = new AtomicLong();

    private final long failedCallThreshold;
    private final long retryThreshold;


    /**
     * Construct a {@code TimedCircuitBreaker}.
     */
    public TunableCircuitBreaker(){
        this(FAILED_CALL_THRESHOLD, RETRY_THRESHOLD);
    }


    /**
     * Construct a {@code TunableCircuitBreaker}.
     *
     * @param failedCallThreshold
     *           The number of calls that must fail before the circuit breaker is opened.
     * @param retryThreshold
     *           The amount of time in milliseconds before the circuit breaker will retry a connection
     */
    public TunableCircuitBreaker(long failedCallThreshold, long retryThreshold){
        this.failedCallThreshold = failedCallThreshold;
        this.retryThreshold      = retryThreshold;
    }


    // Implements CircuitBreaker#callFailed(Exception)
    public void callFailed(T error) {
        if(Status.HALF_OPENED == status()){
            if(status.compareAndSet(Status.HALF_OPENED, Status.OPENED)){
                notifyStatusChange(Status.HALF_OPENED, Status.OPENED);
            }

            nextTryTime.set(System.currentTimeMillis() + getRetryThreshold());
            lastException.compareAndSet(null, error);

        } else if(Status.CLOSED == status()){
            final long failed = failedCalls.incrementAndGet();
            if (failed >= getFailedCallThreshold()) {
                if (status.compareAndSet(Status.CLOSED, Status.OPENED)) {
                    notifyStatusChange(Status.CLOSED, Status.OPENED);
                }

                nextTryTime.set(System.currentTimeMillis() + getRetryThreshold());
                lastException.compareAndSet(null, error);

            }
        }

    }


    // Implements CircuitBreaker#callStarted()

    @SuppressWarnings({"ThrowableResultOfMethodCallIgnored"})
    public void callStarted() throws T {
        if (status() == Status.OPENED) {
            final long currentNextTryTime         = nextTryTime.get();
            final long currentTime                = System.currentTimeMillis();

            if (currentTime < currentNextTryTime){
                blockedCalls.incrementAndGet();
                TunableCircuitBreaker.<T>throwException(lastException.get());
            } else {
                if(status.compareAndSet(Status.OPENED, Status.HALF_OPENED)){
                    avgRecoveryTime.set(currentTime - currentNextTryTime);
                    notifyStatusChange(Status.OPENED, Status.HALF_OPENED);
                }
            }
        }
    }


    // Implements CircuitBreaker#callSucceeded()
    public void callSucceeded() {
        reset();
    }


    /**
     * @return
     *      time at which the cb was able to reset itself.
     */
     public long getAvgRecoveryTime(){
        return status() == Status.CLOSED ? avgRecoveryTime.get() : NO_CLOSED_YET;
     }


    /**
     * @return
     *     The number of calls that must fail before the circuit breaker is opened.
     */
    public long getFailedCallThreshold() {
        return failedCallThreshold;
    }


    /**
     *
     * @return
     *     The amount of time in milliseconds before the circuit breaker will retry a connection
     */
    public long getRetryThreshold() {
            return retryThreshold;
    }

    private static <T extends Exception> T throwException(Throwable t) throws T {
        try {
            @SuppressWarnings({"unchecked"}) // we are casting from some subclass of exception to exception --- bad bad
            final Constructor<T> c = (Constructor<T>) Exception.class.getDeclaredConstructor(Throwable.class);
            return c.newInstance(t);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * notify whichever application (i.e., logging) that needs an update from
     * the circuit breaker.
     * @param oldStatus
     *          old status
     * @param newStatus
     *          new status
     */
    void notifyStatusChange(Status oldStatus, Status newStatus) {
        if((oldStatus == newStatus) && (newStatus == Status.CLOSED)) return;
        System.out.println("Circuit breaker's status has changed from "
                + oldStatus + " to " + newStatus
                + ((Status.CLOSED != newStatus)
                ? ". Next cooling-down time is set to: "
                        + nextTryTime.get() + " milliseconds. Current blocked calls: "
                        + blockedCalls.get()
                : "")
        );
    }


    /**
     *  reset the circuit breaker.
     */
    private void reset() {
        notifyStatusChange(status(), Status.CLOSED);
        status.set(Status.CLOSED);
        failedCalls.set(0L);
        blockedCalls.set(0L);
        lastException.set(null);
    }


    // Implements CircuitBreaker#status()
    public Status status() {
        return status.get();
    }


    @Override
    public String toString() {
        return ToStrings.toString(
                this,
                of("failedCallThreshold", str(this.failedCallThreshold)),
                of("retryThreshold", str(this.retryThreshold)),
                of("status", str(status()))
        );
    }
}
